package org.lds.simple.disruptor.strategy;

import lombok.extern.slf4j.Slf4j;
import org.lds.simple.disruptor.sequence.Sequence;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class DefaultWaitStrategy implements WaitStrategy {
    private ReentrantLock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();

    private final int THREAD_HOLD = 100;
    private final long TIMEOUT = 5 * 1000;

    @Override
    public long consumerWait(Sequence producerSequence, Sequence consumerSequence) {
        int threadHold = 0;
        long timeout = 0;
        while (!allowConsume(producerSequence, consumerSequence)) {
            if (threadHold > THREAD_HOLD) {
                if (timeout == 0) {
                    timeout = System.currentTimeMillis();
                } else {
                    if (System.currentTimeMillis() - timeout > TIMEOUT) {
                        lock.lock();
                        try {
                            condition.await();
                        } catch (Throwable e) {
                            log.error("await fail", e);
                        } finally {
                            lock.unlock();
                        }
                    }
                }
            } else {
                threadHold++;
            }
        }
        return producerSequence.get();
    }

    @Override
    public void notifyConsumer() {
        lock.lock();
        try {
            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }
}
